package com.yanggu.flink.datastream_api.source

import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.streaming.api.functions.source.datagen.{DataGeneratorSource, RandomGenerator}
import org.apache.flink.streaming.api.scala._

/**
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/datagen/
 * 用于模拟数据的生成, 在没有没有合适的source情况下, 可以进行模拟source进行压测或者数据倾斜测试
 */
object SourceFromDataGenerator {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    environment.addSource(new DataGeneratorSource[SensorReading](new RandomGenerator[SensorReading]() {
      override def next() = {
        SensorReading(random.nextHexString(10), System.currentTimeMillis(), random.nextUniform(0.01D, 0.02D))
      }
    }))
      .print()

    environment.execute()
  }

}
